package com.exercise.rpc.server;/**
 * Created by kete on 2018-10-26.
 */

import com.exercise.rpc.common.annotation.RpcService;
import com.exercise.rpc.common.decoder.RpcDecoder;
import com.exercise.rpc.common.encoder.RpcEncoder;
import com.exercise.rpc.common.model.RpcRequest;
import com.exercise.rpc.common.model.RpcResponse;
import com.exercise.rpc.common.registry.ServiceRegistry;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import org.apache.commons.collections4.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.StringUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 作者：kete
 * 创建时间：2018-10-26 9:39
 * 功能描述：
 * 版本：
 */
public class RpcServer implements ApplicationContextAware,InitializingBean {

    private final static Logger logger = LoggerFactory.getLogger(RpcServer.class);

    private String serviceAddress;

    private ServiceRegistry registry;


    private Map<String,Object> handlerMap = new ConcurrentHashMap<>();

    public RpcServer(String serviceAddress) {
        this.serviceAddress = serviceAddress;
    }

    public RpcServer(String serviceAddress, ServiceRegistry registry) {
        this.serviceAddress = serviceAddress;
        this.registry = registry;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        EventLoopGroup bossGrop = new NioEventLoopGroup();
        EventLoopGroup workGrop = new NioEventLoopGroup();

        ServerBootstrap server = new ServerBootstrap();
        server.group(bossGrop,workGrop).
               channel(NioServerSocketChannel.class).
               childHandler(new InitHandler()).
               option(ChannelOption.SO_BACKLOG,128).
               option(ChannelOption.SO_KEEPALIVE,true);

        String[] addressArray = serviceAddress.split(":");

        String ip = addressArray[0];
        int port = Integer.parseInt(addressArray[1]);
        // 启动 RPC 服务器
        ChannelFuture future = server.bind(ip, port).sync();

        // 注册 RPC 服务地址
        if (registry != null) {
            registry.register(serviceAddress);
        }
        logger.debug("server started on port {}", port);
        // 关闭 RPC 服务器
        future.channel().closeFuture().sync();
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String,Object> serviceMap = applicationContext.getBeansWithAnnotation(RpcService.class);

        if(MapUtils.isNotEmpty(serviceMap)){

            for(Object serviceBean : serviceMap.values()){
                RpcService annotation = serviceBean.getClass().getAnnotation(RpcService.class);
                String serviceName = annotation.value().getName();
//                String version = annotation.version();
//                if(!StringUtils.isEmpty(version)){
//                    serviceName = serviceName +"-"+version;
//                }

                handlerMap.put(serviceName,serviceBean);
            }
        }

    }

    private class InitHandler extends ChannelInitializer<SocketChannel>{
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline .addLast(new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 0));
            pipeline.addLast(new RpcEncoder(RpcResponse.class));
            pipeline.addLast(new RpcDecoder(RpcRequest.class));
            pipeline.addLast(new ServiceHandler(handlerMap));
        }
    }
}
